package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2024-05-20-14:48
 * @see 2194550857@qq.com
 */
public class Flink01_ProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                    }
                });
        //实现类继承抽象类，需要重写里面的所有抽象方法
        //如果实现类实现接口，需要重写所有的方法，无论是否是抽象
        waterSensorSingleOutputStreamOperator.process(new ProcessFunction<WaterSensor, String>() {
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                out.collect(value.toString());
            }
        }).print();

        env.execute();
    }
}
